#include "comm/znx_string.h"
#include "comm/znx_inet.h"
#include "event/znx_conn.h"
#include "event/znx_event_loop.h"
#include "event/znx_worker_proc.h"

static znx_bool_t use_et_model = ZNX_FALSE;

void znx_test_server_read_handler(znx_event_t *event)
{
    znx_conn_t *conn = event->data;

    if (event->timer_set && !event->timedout) {
        ZNX_DEF_LOG_FATAL(NULL, "unknow event callback");
    }

    if (event->timedout) {
        ZNX_DEF_LOG_DEBUG(NULL, "test_server timeout event");
        event->timedout = 0;

        // enabled read io
        znx_event_loop_enable_read(conn);
        return;
    }

    znx_result_t    ret;
    do {
        char data[1024];
        ret = znx_conn_read(conn, data, sizeof(data) - 1);
        if (ret > 0) {
            data[(int)ret] = '\0';
            ZNX_DEF_LOG_DEBUG(NULL, "test_server read data: %s", data);
        }
    } while(ret > 0);

    if (ret == ZNX_AGAIN) {
        ZNX_DEF_LOG_DEBUG(NULL, "test_server read eagain");
        znx_event_loop_enable_read(conn);
        return;
    }

    if (ret < 0) {
        ZNX_DEF_LOG_ERROR(NULL, "test_server read errno: %d", ret);
    }

    if (ret == 0) {
        ZNX_DEF_LOG_DEBUG(NULL, "test_server conn closed");
    }

    znx_event_loop_remove_conn(conn);
    znx_conn_close_and_free(conn);

    znx_atomic_int64_add_fetch(&pending_tasks, -1);
}


void znx_test_server_handler(znx_dispatch_event_t *dispatch_event)
{
    int accept_fd = (int)(intptr_t)dispatch_event->data;
    znx_thread_free(dispatch_event);

    znx_conn_t *server_conn = znx_thread_calloc(sizeof(znx_conn_t));
    znx_conn_init(server_conn, ZNX_TCP_CONN,
        use_et_model ? ZNX_CONN_EPOLL_ET : ZNX_CONN_EPOLL_LT);
    server_conn->fd = accept_fd;
    znx_atomic_int64_add_fetch(&server_conns, 1);
    server_conn->server = 1;

    znx_event_loop_start_io(server_conn);
    server_conn->read.handler = znx_test_server_read_handler;
    znx_event_loop_add_timer(&server_conn->read, 100);
}

const char *msg = "abcdefghi\n";

void znx_test_client_close_handler(znx_event_t *event)
{
    znx_conn_t *conn = event->data;
    znx_event_loop_remove_conn(conn);
    znx_conn_close_and_free(conn);
    znx_atomic_int64_add_fetch(&pending_tasks, -1);
}

void znx_test_client_write_handler(znx_event_t *event)
{
    znx_conn_t *conn = event->data;

    if (event->timedout) {
        event->timedout = 0;
        znx_event_loop_enable_write(conn);
        return;
    }

    int step = 0;
    if (conn->data != NULL) {
        step = (int)(intptr_t)conn->data;
    }

    step++;
    conn->data = (void *)(intptr_t)step;

    if (step > 5) {
        event->handler = znx_test_client_close_handler;
        znx_event_loop_disable_write(conn);
        znx_event_loop_add_timer(event, 100);
        return;
    }

    znx_result_t ret;
    ret = znx_conn_write(conn, (void *)msg, strlen(msg));
    if (ret != (int64_t)strlen(msg)) {
        ZNX_DEF_LOG_FATAL(NULL, "send failed, except: %d, but got %d. errno: %d", strlen(msg), ret, -ret);
    }

    znx_event_loop_disable_write(conn);
    znx_event_loop_add_timer(event, 100);
}

void znx_test_client_timer_handler(znx_event_t *event)
{
    znx_conn_t *conn = event->data;

    if (!event->timedout) {
        ZNX_DEF_LOG_FATAL(NULL, "unknow event callback");
    }

    event->timedout = 0;
    conn->write.handler = znx_test_client_write_handler;

    znx_event_loop_enable_write(conn);
}

void znx_test_client_handler(znx_dispatch_event_t *dispatch_event)
{
    int conn_fd = (int)(intptr_t)dispatch_event->data;
    znx_thread_free(dispatch_event);

    znx_conn_t *client_conn = znx_thread_calloc(sizeof(znx_conn_t));
    znx_conn_init(client_conn, ZNX_TCP_CONN,
        use_et_model ? ZNX_CONN_EPOLL_ET : ZNX_CONN_EPOLL_LT);
    client_conn->fd = conn_fd;
    znx_atomic_int64_add_fetch(&client_conns, 1);
    client_conn->client = 1;

    znx_event_loop_start_io(client_conn);
    client_conn->write.handler = znx_test_client_timer_handler;
    znx_event_loop_add_timer(&client_conn->write, 100);
}

void test_conn()
{
    znx_result_t    ret;

    // socket bind listen

    ret = znx_syscall_socket(AF_INET, SOCK_STREAM, 0);
    if (ret < 0) {
        ZNX_DEF_LOG_FATAL(NULL, "socket() failed, errno: %d", -ret);
    }

    int listen_fd = (int)ret;
    int reuseaddr_val = 1;
    ret = znx_syscall_setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &reuseaddr_val, sizeof(int));
    if (ret < 0) {
        ZNX_DEF_LOG_FATAL(NULL, "setsockopt(SO_REUSEADDR) failed, errno: %d", -ret);
    }

    znx_address_t address;
    if (!znx_address_parser(
        (u_char *)"127.0.0.1:8889", strlen("127.0.0.1:8889"), &address, NULL)) {
        ZNX_DEF_LOG_FATAL(NULL, "parser address failed");
    }

    socklen_t sockaddr_len = sizeof(znx_sockaddr_in_t);

    ret = znx_syscall_bind(listen_fd, &address.sockaddr, sockaddr_len);
    if (ret < 0) {
        ZNX_DEF_LOG_FATAL(NULL, "bind() failed, errno: %d", -ret);
    }

    ret = znx_syscall_listen(listen_fd, 64);
    if (ret < 0) {
        ZNX_DEF_LOG_FATAL(NULL, "listen() failed, errno: %d", -ret);
    }

    // socket connect
    ret = znx_syscall_socket(AF_INET, SOCK_STREAM, 0);
    if (ret < 0) {
        ZNX_DEF_LOG_FATAL(NULL, "socket() failed, errno: %d", -ret);
    }

    int conn_fd = (int)ret;
    ret = znx_syscall_connect(conn_fd, &address.sockaddr, sockaddr_len);
    if (ret < 0) {
        ZNX_DEF_LOG_FATAL(NULL, "connect() failed, errno: %d", -ret);
    }

    ret = znx_syscall_accept4(listen_fd, NULL, NULL, SOCK_NONBLOCK|SOCK_CLOEXEC);
    if (ret < 0) {
        ZNX_DEF_LOG_FATAL(NULL, "accept() failed, errno: %d", -ret);
    }

    int accept_fd = (int)ret;

    znx_atomic_int64_add_fetch(&pending_tasks, 2);

    znx_dispatch_event_t *server_dispatch_event = znx_thread_calloc(sizeof(znx_dispatch_event_t));
    server_dispatch_event->handler = znx_test_server_handler;
    server_dispatch_event->data = (void *)(intptr_t)accept_fd;

    znx_dispatch_event_t *client_dispatch_event = znx_thread_calloc(sizeof(znx_dispatch_event_t));
    client_dispatch_event->handler = znx_test_client_handler;
    client_dispatch_event->data = (void *)(intptr_t)conn_fd;

    znx_worker_processor_post_event(client_dispatch_event);
    znx_worker_processor_post_event(server_dispatch_event);
}

int main(int argc, char **argv)
{
    if (argc > 1 && strcmp(argv[1], "et") == 0) {
        use_et_model = ZNX_TRUE;
    }

    if (use_et_model) {
        ZNX_DEF_LOG_DEBUG(NULL, "use ET model");
    } else {
        ZNX_DEF_LOG_DEBUG(NULL, "use LT model");
    }

    // 1. 先调用proc_init
    znx_proc_init();

    // 2. core conf初始化
    znx_core_conf_create_default();
    g_core_conf->worker_threads = 5;
    g_core_conf->tcp_accepter_threads = 1;
    g_core_conf->g_mem_freelists_level = 1;
    g_core_conf->g_aligned_mem_freelists_level = 1;
    g_core_conf->worker_mem_freelist_level = 1;
    g_core_conf->worker_aligned_mem_freelists_level = 1;

    // 3. 初始化日志相关配置
    znx_def_log_init(0, LOG_LEVEL_DEBUG, NULL, LOG_ROLLING_DAY);

    // 4. listener conf初始化
    // 5. http conf 初始化

    // 6. 调用 proc start
    znx_proc_start();

    test_conn();

    for ( ;; ) {
        if (znx_atomic_int64_load(&pending_tasks) == 0) {
            break;
        }

        usleep(1000);
    }

    // 8. 调用proc shutdown.
    znx_proc_shutdown();

    return 0;
}
